package com.atguigu.tingshu.album.service.impl;

import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.cache.TsCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumInfoMapper albumInfoMapper;

    @Autowired
    private AlbumStatMapper albumStatMapper;

    @Autowired
    private AlbumAttributeValueMapper albumAttributeValueMapper;

    @Autowired
    private AlbumAttributeValueService albumAttributeValueService;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    @Override
    @TsCache(prefix = "stat:")
    public AlbumStatVo getAlbumStatVo(Long albumId) {
        //	查询数据
        return albumInfoMapper.selectAlbumStatVo(albumId);
    }

    @Override
    @Transactional
    public void test() {
        AlbumInfo albumInfo = new AlbumInfo();
        albumInfo.setUserId(1l);
        albumInfo.setIncludeTrackCount(5);
        albumInfoMapper.insert(albumInfo);
        test1();
    }

    @Transactional
    public void test1() {
        AlbumStat albumStat = new AlbumStat();
        albumStat.setId(1l);
        albumStat.setStatNum(1);
        int i = 1 / 0;
        //	albumInfoMapper.updateById(albumInfo);
        albumStatMapper.updateById(albumStat);
    }

    @Override
    public List<AlbumInfo> findUserAllAlbumList(Long userId) {
        //	查询album_info
        //	select * from album_info from where user_id = ?;
        //	select id, album_title from album_info from where user_id = ? order by id desc limit 100;
        return albumInfoMapper.selectList(new LambdaQueryWrapper<AlbumInfo>().eq(AlbumInfo::getUserId, userId).select(AlbumInfo::getId, AlbumInfo::getAlbumTitle).orderByDesc(AlbumInfo::getId).last(" limit 100 "));
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateAlbumInfo(Long albumId, AlbumInfoVo albumInfoVo) {
        //  先组成缓存的key
        String key = "album:["+albumId+"]";
        this.redisTemplate.delete(key);
        //	专辑信息；album_info
        //	标签：album_attribute_value
        AlbumInfo albumInfo = new AlbumInfo();
        //	属性拷贝.
        BeanUtils.copyProperties(albumInfoVo, albumInfo);
        albumInfo.setId(albumId);
        albumInfoMapper.updateById(albumInfo);
        //  修改缓存.
        //	album_attribute_value 先删除，再新增.
        albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
        //	获取专辑标签集合.
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        //	判断是否为空.
        if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
            //	循环遍历.
            //			albumAttributeValueVoList.forEach(albumAttributeValueVo -> {
            //				//	创建对象
            //				AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
            //				//	属性拷贝.
            //				BeanUtils.copyProperties(albumAttributeValueVo,albumAttributeValue);
            //				albumAttributeValue.setAlbumId(albumId);
            //				//	多次执行insert.
            //				albumAttributeValueMapper.insert(albumAttributeValue);
            //			});
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
                //	创建对象
                AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
                //	属性拷贝.
                BeanUtils.copyProperties(albumAttributeValueVo, albumAttributeValue);
                albumAttributeValue.setAlbumId(albumId);
                //	多次执行insert.
                return albumAttributeValue;
            }).collect(Collectors.toList());
            //	批量保存。
            albumAttributeValueService.saveBatch(albumAttributeValueList);
        }
        //  睡眠
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        this.redisTemplate.delete(key);
        //	判断当前专辑是否为私密：
        if ("1".equals(albumInfoVo.getIsOpen())) {
            //	专辑上架  album发送消息，发送的内容是什么?--根据消费者处理消息的业务逻辑. albumId;
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
        } else {
            //	专辑下架
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, albumId.toString());
        }
    }

    @Override
    @TsCache(prefix = "album:")
    public AlbumInfo getAlbumInfo(Long albumId) {
        return this.getAlbumInfoDB(albumId);
        //  return getAlbumInfoByRedisson(albumId);
    }

    @Nullable
    private AlbumInfo getAlbumInfoByRedisson(Long albumId) {
        //	声明对象
        AlbumInfo albumInfo = new AlbumInfo();
        //	先查询缓存数据.
        String key = RedisConstant.ALBUM_INFO_PREFIX + albumId;
        try {
            //	从缓存中获取数据
            albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(key);
            //	判断
            if (null == albumInfo) {
                //	上锁：
                String locKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
                //	先使用redis + lua   setnx key value setex key timeout value;  redis2.6.12以后版本；set key value ex timeout nx
                //  Boolean result = this.redisTemplate.opsForValue().setIfAbsent(locKey, token, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
                RLock lock = redissonClient.getLock(locKey);
                //  上锁
                lock.lock();
                //  boolean result = lock.tryLock(RedisConstant.ALBUM_LOCK_EXPIRE_PX1, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
                //	判断是否获取到锁.
                //  业务逻辑
                try {
                    //  再次查询缓存数据，看是否有数据.
                    albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(key);
                    if (null!=albumInfo){
                        //  不为空，直接返回数据.
                        return albumInfo;
                    }
                    //	缓存中没有数据.
                    albumInfo = this.getAlbumInfoDB(albumId);
                    //	判断数据库中是否有数据
                    if (null == albumInfo) {
                        //	缓存一个空数据存储到缓存
                        this.redisTemplate.opsForValue().set(key, new AlbumInfo(), RedisConstant.ALBUM_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
                        //	调用方法删除key
                        return new AlbumInfo();
                    } else {
                        //	数据库中有数据
                        this.redisTemplate.opsForValue().set(key, albumInfo, RedisConstant.ALBUM_TIMEOUT, TimeUnit.DAYS);
                        //	调用方法删除key
                        return albumInfo;
                    }
                } finally {
                    //  解锁
                    lock.unlock();
                }
            }else {
                //	返回缓存数据
                return albumInfo;
            }
        } catch (RuntimeException e) {
            log.error("xxxx"+e.getMessage());
            //	调用发送短信方法. redis 宕机了.....
            e.printStackTrace();
        }
        //	获取专辑详情数据
        return getAlbumInfoDB(albumId);
        //  return getAlbumInfoByRedis(albumId);
    }

    @Nullable
    private AlbumInfo getAlbumInfoByRedis(Long albumId) {
        //	声明对象
        AlbumInfo albumInfo = new AlbumInfo();
        //	先查询缓存数据.
        String key = RedisConstant.ALBUM_INFO_PREFIX + albumId;
        try {
            //	从缓存中获取数据
            albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(key);
            //	判断
            if (null == albumInfo) {
                //	上锁：
                String locKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
                String token = UUID.randomUUID().toString().replaceAll("-", "");
                //	先使用redis + lua
                Boolean result = this.redisTemplate.opsForValue().setIfAbsent(locKey, token, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
                //	判断是否获取到锁.
                if (result) {
                    //	缓存中没有数据.
                    albumInfo = this.getAlbumInfoDB(albumId);
                    //	判断数据库中是否有数据
                    if (null == albumInfo) {
                        //	缓存一个空数据存储到缓存
                        this.redisTemplate.opsForValue().set(key, new AlbumInfo(), RedisConstant.ALBUM_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
                        //	调用方法删除key
                        delLocKey(key, token);
                        return new AlbumInfo();
                    } else {
                        //	数据库中有数据
                        this.redisTemplate.opsForValue().set(key, albumInfo, RedisConstant.ALBUM_TIMEOUT, TimeUnit.DAYS);
                        //	调用方法删除key
                        delLocKey(key, token);
                        return albumInfo;
                    }
                } else {
                    //	等待自旋.
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    return getAlbumInfo(albumId);
                }

            }else {
                //	返回缓存数据
                return albumInfo;
            }
        } catch (RuntimeException e) {
            log.error("xxxx"+e.getMessage());
            //	调用发送短信方法. redis 宕机了.....
            e.printStackTrace();
        }
        //	获取专辑详情数据
        return getAlbumInfoDB(albumId);
    }

    private void delLocKey(String key, String token) {
		//	使用lua脚本解锁.
		String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
				"then\n" +
				"    return redis.call(\"del\",KEYS[1])\n" +
				"else\n" +
				"    return 0\n" +
				"end";
		//  执行lua 脚本删除。
		DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
		redisScript.setScriptText(scriptText);
		redisScript.setResultType(Long.class);
		this.redisTemplate.execute(redisScript, Arrays.asList(key), token);
	}

	@Nullable
    private AlbumInfo getAlbumInfoDB(Long albumId) {
        //	获取专辑对象并回显.
        AlbumInfo albumInfo = albumInfoMapper.selectById(albumId);
        //	获取专辑属性标签集合.
        if (null != albumInfo) {
            //	album_attribute_value condition =album_id
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
            albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
        }
        //	返回数据
        return albumInfo;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void removeAlbumInfo(Long albumId) {
        //	稍微思考一下: 删除有两种方式 ： 一种方式逻辑删除 一种方式物理删除：delete from table; truncate table tname;
		/*album_attribute_value
		album_info
		album_stat*/
        albumInfoMapper.deleteById(albumId);
        //	本质执行的是update sql
        LambdaQueryWrapper<AlbumStat> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(AlbumStat::getAlbumId, albumId);
        albumStatMapper.delete(wrapper);
        albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));

        //	发送消息.
        kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, albumId.toString());
    }

    @Override
    public IPage<AlbumListVo> findUserAlbumPage(Page<AlbumListVo> albumListVoPage, AlbumInfoQuery albumInfoQuery) {
        //	调用mapper层方法.
        return albumInfoMapper.selectUserAlbumPage(albumListVoPage, albumInfoQuery);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveAlbumInfo(AlbumInfoVo albumInfoVo) {
        try {
            //	先保存专辑信息.
            AlbumInfo albumInfo = new AlbumInfo();
            //	属性拷贝：
            BeanUtils.copyProperties(albumInfoVo, albumInfo);
            //	user_id
            albumInfo.setUserId(1l);
            //	status
            albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
            //	判断当前专辑的类型.
            if (!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(albumInfoVo.getPayType())) {
                albumInfo.setTracksForFree(5);
            }
            //	数据保存.
            albumInfoMapper.insert(albumInfo);

            //	保存专辑标签表：
            List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
            if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
                //	循环遍历》
                albumAttributeValueVoList.forEach(albumAttributeValueVo -> {
                    //	给这个对象进行赋值
                    AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
                    //	给专辑Id 进行赋值. 因为先执行了insert 方法，然后通过。  @TableId(type = IdType.AUTO) 配置了获取主键自增.
                    albumAttributeValue.setAlbumId(albumInfo.getId());
                    albumAttributeValue.setAttributeId(albumAttributeValueVo.getAttributeId());
                    albumAttributeValue.setValueId(albumAttributeValueVo.getValueId());
                    //	保存数据. 单独执行5次insert into 方法.
                    albumAttributeValueMapper.insert(albumAttributeValue);
                });
                //	批量保存...
            }
            //	初始化专辑统计信息.
            this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_PLAY);
            this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_SUBSCRIBE);
            this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_BROWSE);
            this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_COMMENT);

            //	-- 审核通过之后，才能上架. 数据万象--异步审核.
            //	判断当前专辑是否为私密：
            if ("1".equals(albumInfoVo.getIsOpen())) {
                //	专辑上架  album发送消息，发送的内容是什么?--根据消费者处理消息的业务逻辑. albumId;
                kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumInfo.getId().toString());
            } else {
                //	专辑下架
                kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, albumInfo.getId().toString());
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void saveAlbumStat(Long albumId, String statPlay) {
        //	创建对象
        AlbumStat albumStat = new AlbumStat();
        albumStat.setAlbumId(albumId);
        albumStat.setStatType(statPlay);
        albumStat.setStatNum(0);
        albumStatMapper.insert(albumStat);
    }
}
